package com.csap.framework.integration.client.boot.autoconfigure;

import com.aliyun.openservices.ons.api.MessageListener;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.bean.ConsumerBean;
import com.aliyun.openservices.ons.api.bean.Subscription;
import com.csap.framework.integration.RocketmqConfig;
import com.csap.framework.integration.boot.autoconfigure.RocketmqAutoConfiguration;
import com.csap.framework.integration.consumer.EnableRocketmqBean;
import com.csap.framework.integration.listener.SimpleMessageListener;
import org.springframework.boot.autoconfigure.ImportAutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

/**
 * @author yangchengfu
 * @description
 * @dataTime 2020年-10月-12日 15:45:00
 **/
@Configuration
@ComponentScan
@ImportAutoConfiguration(RocketmqAutoConfiguration.class)
@ConditionalOnProperty(prefix = RocketmqConfig.prefix + ".consumer", name = "enabled", havingValue = "true")
public class RocketmqClientAutoConfiguration {

    @Bean
    @ConditionalOnBean({EnableRocketmqBean.class})
    public SimpleMessageListener simpleMessageListener() {
        return new SimpleMessageListener();
    }

    @Bean(initMethod = "start", destroyMethod = "shutdown")
    @ConditionalOnBean({EnableRocketmqBean.class, RocketmqConfig.class})
    public ConsumerBean consumerBean(RocketmqConfig rocketmqConfig, EnableRocketmqBean enableRocketmqBean, ApplicationContext applicationContext) {
        ConsumerBean consumerBean = new ConsumerBean();
        //配置文件
        Properties properties = rocketmqConfig.getConsumerProperties();
        properties.setProperty(PropertyKeyConst.GROUP_ID, rocketmqConfig.getConsumer().getGroupId());
        //将消费者线程数固定为20个 20为默认值
        properties.setProperty(PropertyKeyConst.ConsumeThreadNums, rocketmqConfig.getConsumer().getConsumeThreadNums().toString());
        consumerBean.setProperties(properties);
        if (enableRocketmqBean != null) {
            //订阅关系
            Map<Subscription, MessageListener> subscriptionTable = new HashMap<>(16);
            if (!rocketmqConfig.getConsumer().getEnabledMessageFilter()) {
                enableRocketmqBean.getRocketmqBeans().forEach((k, v) -> {
                    Subscription subscription = new Subscription();
                    subscription.setTopic(v.getTopic());
                    subscription.setExpression(v.getTag());
                    subscriptionTable.put(subscription, applicationContext.getBean(v.getAClass()));
                });
            } else {
                Subscription subscription = new Subscription();
                subscription.setTopic(rocketmqConfig.getConsumer().getTopic());
                subscription.setExpression(rocketmqConfig.getConsumer().getTag());
                subscriptionTable.put(subscription, applicationContext.getBean(SimpleMessageListener.class));
            }
            //订阅多个topic如上面设置
            consumerBean.setSubscriptionTable(subscriptionTable);
        }
        return consumerBean;
    }

}
